package cn.itcast.flink.start;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

/**
 * * 基于Flink SQL实现数据离线分析：加载文件系统中文本文件数据（csv格式数据），
 * 封装到Table表中，进行数据分析处理
 *
 * @author lilulu
 * @date 2023-04-10 15:53
 */
public class FlinkSqlDemoV1 {
    public static void main(String[] args) {
        /**
         * 1、创建表执行环境
         * 2、创建输入表
         * 3、创建输出表
         * 4、查询分析表数据
         * 5、将结果数据输出结果表
         */
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inBatchMode().useBlinkPlanner().build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);

        tableEnv.executeSql("CREATE TABLE tbl_ratings(\n" +
                "user_id STRING,\n" +
                "movie_id STRING,\n" +
                "rating DOUBLE,\n" +
                "ts BIGINT\n" +
                ") WITH (\n" +
                "'connector' = 'filesystem',\n" +
                "'path' = 'datas/ratings.data',\n" +
                "'format' = 'csv',\n" +
                "'csv.field-delimiter' = '\\t',\n" +
                "'csv.ignore-parse-errors' = 'true'\n" +
                ")");

//        分析电影数据，获取最受欢迎10部电影：Top10电影
//        sql写法
        TableResult tableResult = tableEnv.executeSql("select * from tbl_ratings limit 10");
        tableResult.print();
        System.out.println("============");
//        tableApi写法
        tableEnv.from("tbl_ratings").limit(10).execute().print();

    }
}
